package com.mjf.day2

import org.apache.flink.streaming.api.scala._

/**
 * union：
 *  将多个流【必须是同一事件类型的流】合并为一个流，相当于SQL中的union all
 */
object UnionExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val bj: DataStream[SensorReading] = env.addSource(new SensorSource).filter(_.id.equals("sensor_1"))
    val sh: DataStream[SensorReading] = env.addSource(new SensorSource).filter(_.id.equals("sensor_2"))
    val sz: DataStream[SensorReading] = env.addSource(new SensorSource).filter(_.id.equals("sensor_3"))

    val union: DataStream[SensorReading] = bj.union(sh, sz)

    union.print()

    env.execute("UnionExample")
  }
}
